package com.wudl.flink.sql;

import com.wudl.flink.bean.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.table.planner.plan.nodes.calcite.LogicalWindowTableAggregate;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;


/**
 * @ClassName : Flink_Sql_Function_UDTFA
 * @Description : Flink自定义udtfa 函数
 * @Author :wudl
 * @Date: 2021-08-18 23:55
 */

public class Flink_Sql_Function_UDTFA {
    public static void main(String[] args) throws Exception {


        //1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //2. 读取端口中的数据并且转化为javaBean
        SingleOutputStreamOperator<WaterSensor> waterSensorDs = env.socketTextStream("192.168.1.161", 9999)
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                });

        // 3. 讲流 转化为动态表
        Table table = tableEnv.fromDataStream(waterSensorDs);


        // 5. 先注册在使用
        tableEnv.createTemporarySystemFunction("Top2", Top2.class);
        //5.1 使用table api 实现的方式
        table.groupBy($("id"))
                .flatAggregate(call("top2",$("vc")).as("top","rank"))
                .select($("id"),$("top"),$("rank"))
                .execute()
                .print();
        // 5.2 采用sql 的写法

        //5. 执行任务
        env.execute();


    }

    // 自定义函数类Udtf 求平均数
    public static class Top2 extends TableAggregateFunction<Tuple2<Integer,String>,VcTop2> {


        public void accumulate(VcTop2 acc, Integer value) {

            if (value > acc.getTopOne()) {
                acc.setTopTwo(acc.getTopOne());
                acc.setTopOne(value);
            } else if (value > acc.getTopTwo()) {
                acc.setTopTwo(value);
            }
        }

        public void emitValue(VcTop2 acc, Collector<org.apache.flink.api.java.tuple.Tuple2<Integer, String>> out) {
            out.collect(new org.apache.flink.api.java.tuple.Tuple2<>(acc.getTopOne(), "Top1"));
            if (acc.getTopTwo() > Integer.MIN_VALUE) {
                out.collect(new org.apache.flink.api.java.tuple.Tuple2<>(acc.getTopTwo(), "Top2"));
            }
        }

        @Override
        public VcTop2 createAccumulator() {
            return new VcTop2(Integer.MIN_VALUE, Integer.MIN_VALUE);
        }
    }

}
